package com.test.cn.spark.streaming.graph
import org.apache.spark.graphx.{Edge, EdgeTriplet, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object GraphXDemo {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    //定义定点
    val vertexArray: Array[(VertexId, String)] =
      Array((1, "SFO"), (2, "ORD"), (3, "DFW"))
    val vertexRDD: RDD[(VertexId, String)] = sc.makeRDD(vertexArray)

    //定义边
    val edgeArray: Array[Edge[Int]] =
      Array(Edge(1L, 2L, 1800), Edge(2L, 3L, 800), Edge(3L, 1L, 1400))
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(edgeArray)

    //定义图
    val graph: Graph[String, Int] = Graph(vertexRDD, edgeRDD)
    //    求所有的顶点
    println("*************求所有的顶点******************")
    graph.vertices.foreach(println)

    //    求所有的边
    println("*************求所有的边******************")
    graph.edges.foreach(println)
//    求所有的triplets
    println("*************求所有的triplets******************")
    graph.triplets.foreach(println)
//    求顶点数
    println("*************求顶点数******************")
    println(graph.vertices.count())
//    求边数
    println("*************求边数******************")
    println(graph.edges.count())
//    求机场距离大于1000的有几个，有哪些按所有机场之间的距离排序（降序），输出结果
    println("*************求机场距离大于1000的有几个******************")
    val filterRDD: RDD[EdgeTriplet[String, Int]] = graph.triplets.filter((triplet)=> triplet.attr>800)
    println(filterRDD.count())
    filterRDD.sortBy(_.attr).foreach(println)


    sc.stop()

  }

}
